package prictice.transaction1;

import org.apache.storm.coordination.BatchOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseTransactionalBolt;
import org.apache.storm.transactional.ICommitter;
import org.apache.storm.transactional.TransactionAttempt;
import org.apache.storm.tuple.Tuple;

import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;

/**
 * @Package prictice.transaction1
 * @Description:
 * @Author elwyn
 * @Date 2017/9/16 20:22
 * @Email elonyong@163.com
 */
public class MyCommitter extends BaseTransactionalBolt implements ICommitter {
    private static final long serialVersionUID = -1502810353308340730L;
    public static final String GLOBAL_KEY = "GLOBAL_KEY";
    public static Map<String, DbValue> dbMap = new HashMap<>();
    int sum = 0;
    TransactionAttempt id;
    BatchOutputCollector batchOutputCollector;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, BatchOutputCollector batchOutputCollector, TransactionAttempt transactionAttempt) {
        this.batchOutputCollector = batchOutputCollector;
        this.id = transactionAttempt;
    }

    @Override
    public void execute(Tuple tuple) {
        sum += tuple.getInteger(1);
    }

    @Override
    public void finishBatch() {
        DbValue dbValue = dbMap.get(GLOBAL_KEY);
        DbValue newValue;
        if (dbValue == null || !dbValue.txId.equals(id.getTransactionId())) {
            //更新数据库
            newValue = new DbValue();
            newValue.txId = id.getTransactionId();
            if (dbValue == null) {
                newValue.count = sum;
            } else {
                newValue.count = dbValue.count + sum;
            }
            dbMap.put(GLOBAL_KEY, newValue);
        } else {
            newValue = dbValue;
        }
        System.out.println("total============================:" + dbMap.get(GLOBAL_KEY).count);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    static class DbValue {
        BigInteger txId;
        int count = 0;
    }
}
